package org.zjt.flink;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.IterativeStream;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Objects;

/**
 * Description:
 *
 * @author juntao.zhang
 * Date: 2018-09-30 上午11:46
 * @see
 */
public class BasicOperationJob {

    public static void main(String[] args) throws Exception {
        LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        env.setBufferTimeout(300);

        DataStream<Long> someIntegers = env.generateSequence(0, 1000);

        IterativeStream<Long> iteration = someIntegers.iterate();

        DataStream<Long> minusOne = iteration.map(new MapFunction<Long, Long>() {
            @Override
            public Long map(Long value) throws Exception {
                return value - 1 ;
            }
        }).setBufferTimeout(200);

        DataStream<Long> stillGreaterThanZero = minusOne.filter(new FilterFunction<Long>() {
            @Override
            public boolean filter(Long value) throws Exception {
                System.out.println("stillGreaterThanZero" + value);
                return (value > 0);
            }
        });

        iteration.closeWith(stillGreaterThanZero);



        DataStream<Long> lessThanZero = minusOne.filter(new FilterFunction<Long>() {
            @Override
            public boolean filter(Long value) throws Exception {
                System.out.println("lessThanZero" + value);
                return (value <= 0);
            }
        });

        env.execute("SSSSSSS");
    }


    public static class MyMapper implements MapFunction<Long,Long> {

        @Override
        public Long map(Long value) throws Exception {
            return Objects.isNull(value) ? 0 : value + 1;
        }
    }
}
